package com.gjy.pulsar.admin;

import com.gjy.pulsar.util.PulsarUtil;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;

import java.util.List;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-06 14:21:08
 */
public class TopicsApi {

    private static final PulsarAdmin ADMIN;

    static {
        try {
            ADMIN = PulsarAdmin.builder().serviceHttpUrl(PulsarUtil.ADMIN_URL)
                    .build();
        } catch (PulsarClientException e) {
            throw new RuntimeException(e);
        }
    }

    public static Message<byte[]> getMessageById(String topic, Integer ledgerId, Integer entryId) {
        /*String topic = "persistent://my-tenant/my-namespace/my-topic";
        long ledgerId = 10;
        long entryId = 10;*/
        try {
            return ADMIN.topics().getMessageById(topic, ledgerId, entryId);
        } catch (PulsarAdminException e) {
            throw new RuntimeException(e);
        }

    }

    public static List<Message<byte[]>> topicsPeekMessage(String topic, String subName, Integer numMessages) {
        /*String topic = "persistent://public/default/my-topic";
        String subName = "my-subscription";
        int numMessages = 1;*/
        try {
            return ADMIN.topics().peekMessages(topic, subName, numMessages);
        } catch (PulsarAdminException e) {
            throw new RuntimeException(e);
        }

    }

    public static List<String> topicsList(String tenant, String namespace) {
        try {
            return ADMIN.topics().getList(tenant + "/" + namespace);
        } catch (PulsarAdminException e) {
            throw new RuntimeException(e);
        }

    }

}
